/*
 * Singularity Engine
 * Copyright (C) 2010 Vyacheslav Vorobyov, All rights reserved.
 * See LICENSE for full license information.
 */
package org.singularity.io.util;

import java.io.IOException;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.singularity.io.CloseablePipe;

/**
 * Asynchronizer pipe. This pipe intended to be used as 
 * a wrapper for other pipes to make them asynchronous, 
 * i.e. it may be considered as an asynchronous bridge 
 * between pipes in a pipe line.   
 * 
 * @author vjache
 *
 */
public class APipe<T> extends TPipe<T,T> implements RequireExecutor {
	private class Flusher implements Runnable
	{
		public volatile boolean _active = false;
		@Override
		public void run() 
		{
			_active = true;
			try
			{
				T item = null;
				while((item = _queue.poll(200, TimeUnit.MILLISECONDS))!=null)
				{
					getNext().push(item);
				}
			}
			catch (Exception e) {
				onException(e);
			}
			finally
			{
				_active = false;
			}
		}
	}

	private 	LinkedBlockingQueue<T> 	_queue;
	private 	Flusher 				_flusher = new Flusher();
	protected 	Executor 				_exec;

	/**
	 * 
	 */
	public APipe() {}

	/**
	 * @param aNext
	 */
	public APipe(CloseablePipe<T> aNext, Executor aExec, int aQueueSize) {
		super(aNext);
		setExecutor(aExec);
		setQueueMaxSize(aQueueSize);
	}

	public void setQueueMaxSize(int aSize)
	{_queue = new LinkedBlockingQueue<T>(aSize);}
	
	protected final LinkedBlockingQueue<T> getQueue()
	{return _queue;}

	public void setExecutor(Executor aExec) 
	{_exec = aExec;}
	
	/**
	 * @return the exec
	 */
	protected final Executor getExecutor() 
	{return _exec;}

	/* (non-Javadoc)
	 * @see org.singularity.io.Pipe#flush()
	 */
	@Override
	public synchronized void flush() throws IOException {
		//Wait if flusher active
		while(_flusher._active)
		{
			hang();
		}
		//Ensure queue exhausted
		if(_queue.size()>0)
		{
			//If not then relaunch flusher
			_exec.execute(_flusher);
			//Wait queue exhausted state
			while(_queue.size()>0)
			{
				hang();
			}
		}
		super.flush();
	}

	private void hang() throws IOException {
		try 
		{Thread.sleep(100);} 
		catch (InterruptedException e) 
		{throw new IOException(e);}
	}

	/* (non-Javadoc)
	 * @see org.singularity.io.Pipe#push(java.lang.Object)
	 */
	@Override
	public synchronized void push(T aItem) throws IOException {
		try 
		{
			_queue.put(aItem);
			ensureFlusherActive();
		}
		catch (InterruptedException e) 
		{throw new IOException(e);}
	}
	
	@Override
	public synchronized void close() throws IOException {
		super.close();
		_queue = null;
	}
	
	private void ensureFlusherActive() throws IOException {
		if(!_flusher._active)
		{
			_exec.execute(_flusher);
// BUGFIX: Bug prone code disabled
//			while(!_flusher._active)
//			{hang();}
		}
	}

	protected void onException(Exception e) 
	{System.err.println(e);}

}
